package com.gitee.huanminabc.jcommon.multithreading;

import com.gitee.huanminabc.jcommon.multithreading.executor.ThreadFactoryUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * @Description 功能:
 * 1. 延迟一段时间段, 然后把时间内的任务合并成1次操作来处理, 如果一次来的数据太多了,超过阈值,那么就会直接创建新任务先进行处理,不等待下一个时间段
 * 2. 如果在一批数据处理过程出现异常,支持数据重新放入队列重新消费
 * 3. 支持三种模式: 时间优先,数据优先,平级
 * 注意: 数据都是存储到内存中的, 本机宕机数据丢失
 * @Author huanmin
 * @Date 2024/8/1 上午10:47
 */
@Slf4j
public class MergeDelayTimeTask<T> {
    //优先级枚举
    public enum ThreadMotiveEnum {
        //时间延迟优先, 意思不管数据多少,只要到了时间就执行, 不走阈值逻辑
        TIME,
        //数据阈值优先, 意思是数据到了阈值就执行, 不走时间延迟逻辑
        DATA,
        //平级 , 谁先到谁先执行
        OTHER
    }

    //容器枚举只能支持List和Set
    public enum ContainerEnum {
        LIST,
        SET
    }


    private ThreadMotiveEnum threadMotiveEnum = ThreadMotiveEnum.OTHER;

    /**
     * 将多长时间的多次操作合并成1次，单位：毫秒
     */
    private final int intervalMillisecond;
    /**
     * 每次处理多少条数据
     */
    private final int perBatchCount;
    //阈值
    private final int thresholdCount;


    private final Consumer<Collection<T>> batchBizMethod;
    //因为在这个逻辑中是按照时间段来处理的,没有啥并发,基本是窜行的
    private final ScheduledExecutorService scheduledThreadPool;

    //容器
    private final ContainerEnum containerEnum;

    private final AtomicBoolean triggerScheduleNeeded = new AtomicBoolean(true);
    //任务
    private final Map<String, T> task = new ConcurrentHashMap<>();
    //队列记录任务id
    private final Queue<String> queue = new ConcurrentLinkedQueue<>();
    //最大并发数,防止一直重试,导致创建大量的线程,最后会导致系统崩溃了
    private int maxConcurrentCount;
    //是否阻塞
    private final AtomicBoolean isBlock = new AtomicBoolean(false);

    //当前并发数
    private final AtomicInteger currentConcurrentCount = new AtomicInteger(0);

    public void setMaxConcurrentCount(int maxConcurrentCount) {
        this.maxConcurrentCount = maxConcurrentCount;
    }

    //创建容器
    private Collection<T> createContainer() {
        //判断类型是list还是set
        if (containerEnum == ContainerEnum.LIST) {
            return new ArrayList<>();
        } else {
            return new HashSet<>();
        }
    }

    //任务放入到队列中
    private void addTask(T t) {
        String code = t.hashCode() + UUID.randomUUID().toString();
        task.put(code, t);
        queue.add(code);
    }


    //需要阈值和时间, 谁先到谁先执行
    public MergeDelayTimeTask(String taskName, int intervalSecond, int perBatchCount, int thresholdCount, ContainerEnum containerEnum, Consumer<Collection<T>> batchBizMethod) {
        this(taskName, intervalSecond * 1000, perBatchCount, thresholdCount, 1000, ThreadMotiveEnum.OTHER, containerEnum, batchBizMethod);
    }

    //不需要阈值, 满足时间就批量处理
    //延迟时间相关操作基本都是秒级别以上的,所以这里的单位是秒
    public MergeDelayTimeTask(String taskName, int intervalSecond, int perBatchCount, ContainerEnum containerEnum, Consumer<Collection<T>> batchBizMethod) {
        this(taskName, intervalSecond * 1000, perBatchCount, -1, 1000, ThreadMotiveEnum.TIME, containerEnum, batchBizMethod);
    }

    //不需要时间, 当满足阈值就批量处理
    public MergeDelayTimeTask(Consumer<Collection<T>> batchBizMethod, String taskName, int perBatchCount, int thresholdCount, ContainerEnum containerEnum) {
        this(taskName, -1, perBatchCount, thresholdCount, 1000, ThreadMotiveEnum.DATA, containerEnum, batchBizMethod);
    }


    /**
     * 初始化
     *
     * @param intervalMillisecond 时间间隔,单位毫秒
     * @param perBatchCount       每次处理多少条数据, 前提是在这个时间段内有这么多数据,如果没有这么多数据,则取出所有数据,如果有超过这么多数据,则在下一个时间段继续处理
     * @param thresholdCount      阈值, 如果队列中的数据总长度,超过这个值,则直接创建新任务先进行处理,不等待下一个时间段
     * @param maxConcurrentCount  最大并发数
     * @param threadMotiveEnum    优先级枚举
     * @param containerEnum       容器类型 ,如果是Set需要对象实现equals和hashCode方法也可以直接使用注解@EqualsAndHashCode
     * @param batchBizMethod      批次处理逻辑代码
     */
    public MergeDelayTimeTask(String taskName, int intervalMillisecond, int perBatchCount, int thresholdCount, int maxConcurrentCount, ThreadMotiveEnum threadMotiveEnum, ContainerEnum containerEnum, Consumer<Collection<T>> batchBizMethod) {
        intervalMillisecond = Math.max(intervalMillisecond, 1);
        perBatchCount = Math.max(perBatchCount, 1);
        thresholdCount = Math.max(thresholdCount, 1);
        maxConcurrentCount = Math.max(maxConcurrentCount, 1);

        this.scheduledThreadPool = Executors.newScheduledThreadPool(3, ThreadFactoryUtil.createThreadFactory(taskName));
        this.intervalMillisecond = intervalMillisecond;
        this.perBatchCount = perBatchCount;
        this.batchBizMethod = batchBizMethod;
        this.thresholdCount = thresholdCount;
        this.maxConcurrentCount = maxConcurrentCount;
        this.threadMotiveEnum = threadMotiveEnum;
        this.containerEnum = containerEnum;
    }

    //生产者
    public void produce(T t) {
        //将任务放入队列
        addTask(t);
        //判断是否需要触发延迟任务
        if (threadMotiveEnum == ThreadMotiveEnum.TIME || threadMotiveEnum == ThreadMotiveEnum.OTHER) {
            //判断是否需要触发新延迟任务
            if (triggerScheduleNeeded.compareAndSet(true, false)) {
                //启动一个新任务延迟intervalSecond秒后执行
                scheduledThreadPool.schedule(this::timeConsumeMsg, intervalMillisecond, TimeUnit.MILLISECONDS);
            }
        }

        //判断是否触发阈值逻辑
        if (threadMotiveEnum == ThreadMotiveEnum.DATA || threadMotiveEnum == ThreadMotiveEnum.OTHER) {
            //判断当前并发数是否超过最大并发数,防止不断触发(阈值)
            if (currentConcurrentCount.get() >= maxConcurrentCount) {
                log.warn("并发限制: taskName:{} , currentConcurrentCount:{} , maxConcurrentCount:{} , queue-size:{}  ", Thread.currentThread().getName(), currentConcurrentCount.get(), maxConcurrentCount, queue.size());
                return;
            }
            //判断是否需要触发新任务(阈值)
            if (queue.size() >= thresholdCount) {
                synchronized (this) {
                    if (queue.size() >= thresholdCount) {
                        if (threadMotiveEnum == ThreadMotiveEnum.DATA) {
                            //计算需要消费多少次
                            int count = queue.size() / perBatchCount;
                            for (int i = 0; i < count; i++) {
                                thresholdConsumeMsg();
                            }
                        } else {
                            //时间优先,如果阈值满足,那么就一直消费到小于阈值
                            while (queue.size() >= thresholdCount) {
                                thresholdConsumeMsg();
                            }
                        }
                    }

                }
            }
        }
    }

    private void thresholdConsumeMsg() {
        //获取队列中这个时间段内需要处理的数据 每次取出perBatchCount条数据限制处理的数据量
        Collection<T> collection = this.createContainer();
        List<String> codes = new ArrayList<>();
        for (int i = 0; i < perBatchCount; i++) {
            String poll = queue.poll();
            if (poll == null) {
                break;
            }
            codes.add(poll);
            T t1 = task.get(poll);
            if (t1 == null) {
                break;
            }
            collection.add(t1);
        }
        scheduledThreadPool.schedule(() -> {
            currentConcurrentCount.incrementAndGet();
            //消费
            try {
                batchBizMethod.accept(collection);
                //将消费后的数据从任务中移除
                codes.forEach(task::remove);
            } catch (Exception e) {
                //如果消费失败,则重新放入队列
                queue.addAll(codes);
                log.error("taskName:{}, queue-size:{},currentSize:{} task:{}  error", Thread.currentThread().getName(), queue.size(), currentConcurrentCount.get(), getTaskParams(codes), e);
            }
            currentConcurrentCount.decrementAndGet();
            //如果出现阻塞,判断是否需要唤醒
            if (isBlock.get() && currentConcurrentCount.get() < maxConcurrentCount) {
                isBlock.set(false);
                synchronized (this) {
                    this.notifyAll();
                }
            }
        }, 0, TimeUnit.MILLISECONDS);
    }


    private void timeConsumeMsg() {
        //判断当前并发数是否超过最大并发数,防止不断创建定时任务
        if (currentConcurrentCount.get() >= maxConcurrentCount) {
            synchronized (this) {
                log.warn("并发限制: taskName:{} ,  currentConcurrentCount:{} , maxConcurrentCount:{} , queue-size:{}  ", Thread.currentThread().getName(), currentConcurrentCount.get(), maxConcurrentCount, queue.size());
                isBlock.set(true);
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        currentConcurrentCount.incrementAndGet();
        //设置为true，保证永远都有新任务去处理数据
        triggerScheduleNeeded.set(true);
        //获取队列长度
        long size = queue.size();
        if (size == 0) {
            //消费--发现无任何数据可供消费
            return;
        }
        //获取队列中这个时间段内需要处理的数据 每次取出perBatchCount条数据限制处理的数据量
        Collection<T> collection = this.createContainer();
        List<String> codes = new ArrayList<>();
        for (int i = 0; i < perBatchCount; i++) {
            String poll = queue.poll();
            if (poll == null) {
                break;
            }
            codes.add(poll);
            T t = task.get(poll);
            if (t == null) {
                break;
            }
            collection.add(t);
        }

        if (!queue.isEmpty()) {
            //说明还有数据未处理，移交到下一个时间段继续触发新任务,  如果这里不这样做,那么就需要等待下次produce方法调用才能继续消费
            scheduledThreadPool.schedule(this::timeConsumeMsg, intervalMillisecond, TimeUnit.MILLISECONDS);
        }

        //消费
        try {
            batchBizMethod.accept(collection);
            //将消费后的数据从任务中移除
            codes.forEach(task::remove);

        } catch (Exception e) {
            //如果消费失败,则重新放入队列
            queue.addAll(codes);
            //判断当前线程数,如果小于等于1那么就需要创建新的来消费失败的,不然可能会一直没人消费, 只有时间消费的需要这个做因为是滚动的
            if (currentConcurrentCount.get() <= 1) {
                scheduledThreadPool.schedule(this::timeConsumeMsg, intervalMillisecond, TimeUnit.MILLISECONDS);
            }
            log.error("taskName:{}, queue-size:{},currentSize:{} task:{}  error", Thread.currentThread().getName(), queue.size(), currentConcurrentCount.get(), getTaskParams(codes), e);
        }
        currentConcurrentCount.decrementAndGet();
        //如果出现阻塞,判断是否需要唤醒
        if (isBlock.get() && currentConcurrentCount.get() < maxConcurrentCount) {
            isBlock.set(false);
            synchronized (this) {
                this.notifyAll();
            }
        }
    }

    //通过任务id取出实际的任务参数,根据集合类型返回
    public Collection<T> getTaskParams(List<String> code) {
        Collection<T> list =  this.createContainer();
        code.forEach(c -> {
            T t = task.get(c);
            if (t != null) {
                list.add(t);
            }
        });
        return list;
    }


}
